import java.io.BufferedReader;
import java.io.File;
import java.io.FileOutputStream;
import java.io.InputStreamReader;
import java.util.GregorianCalendar;

public class MaquinasExecutamQualquerTarefa {
	private static File diretorioBroker = new File("/home/isabellylr/broker");
	private static String diretorioArquivos = "/home/isabellylr/workspace/Twitter";
	private static int NUMERO_DE_MAQUINAS, INDICE_FILTRAGEM, INDICE_CONTAGEM;
	private static String[] maquinas;
	private static long inicio;
	
	public static void main(String[] args) {	
		try {
			iniciarConstantes();
			inicio = new GregorianCalendar().getTimeInMillis();
			executarTarefas();
			imprimeTempo();
		} catch (Exception ex) {
			System.err.println(ex.getMessage());
		}
	}
	
	private static void iniciarConstantes() {
		NUMERO_DE_MAQUINAS = 4;
		INDICE_FILTRAGEM = 1;
		INDICE_CONTAGEM = 1;
		maquinas = new String[NUMERO_DE_MAQUINAS];
	}
	
	private static void executarTarefas() throws Exception {
		System.out.println("----");
		for (int i = 0; i < NUMERO_DE_MAQUINAS; i++) {
			System.out.println("--\nMaquina" + i + ": " + maquinas[i] + "\n--");
			if (maquinas[i] == null || terminouJob(maquinas[i])) {
				System.out.println("Terminou Job: " + maquinas[i]);
				if ((new File("Files/TweetsFiltrados" + INDICE_CONTAGEM + ".txt")).exists()) {
					criaContadorDeOcorrenciaJDF();
					executarJob("ContadorDeOcorrencia" + INDICE_CONTAGEM);
					maquinas[i] = "ContadorDeOcorrencia" + INDICE_CONTAGEM;
					INDICE_CONTAGEM++;
				} else if ((new File("Files/Tweets" + INDICE_FILTRAGEM + ".txt")).exists()) {
					criaFiltroJDF();
					executarJob("Filtro" + INDICE_FILTRAGEM);
					maquinas[i] = "Filtro" + INDICE_FILTRAGEM;
					INDICE_FILTRAGEM++;
				} 
			}
		}
		System.out.println("----");
		System.out.println("Tempo (em milisegundos):" + (new GregorianCalendar().getTimeInMillis() - inicio));
		if (precisaRepetir()) {
			executarTarefas();
		}
	}
	
	private static void criaFiltroJDF() throws Exception {
		FileOutputStream filtroJDF = new FileOutputStream("Jobs/Filtro" + INDICE_FILTRAGEM + ".jdf");
		filtroJDF.write(("job :\nlabel 	: Filtro" + INDICE_FILTRAGEM + "\n\ntask :\n").getBytes());
		filtroJDF.write(("init 	: put " + diretorioArquivos + "/Files/Tweets" + INDICE_FILTRAGEM + ".txt Tweets" + INDICE_FILTRAGEM + ".txt\n").getBytes());
		filtroJDF.write(("put " + diretorioArquivos + "/bin/Filtro.class Filtro.class\n").getBytes());
		filtroJDF.write(("remote 	: java -cp . Filtro Tweets" + INDICE_FILTRAGEM + ".txt TweetsFiltrados" + INDICE_FILTRAGEM + ".txt\n").getBytes());
		filtroJDF.write(("final   : get TweetsFiltrados" + INDICE_FILTRAGEM + ".txt " + diretorioArquivos + "/Files/TweetsFiltrados" + INDICE_FILTRAGEM + ".txt").getBytes());
	}
	
	private static void criaContadorDeOcorrenciaJDF() throws Exception {
		FileOutputStream ocorrenciaJDF = new FileOutputStream("Jobs/ContadorDeOcorrencia" + INDICE_CONTAGEM + ".jdf");
		ocorrenciaJDF.write(("job :\nlabel 	: ContadorDeOcorrencia" + INDICE_CONTAGEM + "\n\ntask :\n").getBytes());
		ocorrenciaJDF.write(("init 	: put " + diretorioArquivos + "/Files/TweetsFiltrados" + INDICE_CONTAGEM + ".txt TweetsFiltrados" + INDICE_CONTAGEM + ".txt\n").getBytes());
		ocorrenciaJDF.write(("put " + diretorioArquivos + "/bin/ContadorDeOcorrencia.class ContadorDeOcorrencia.class\n").getBytes());
		ocorrenciaJDF.write(("remote 	: java -cp . ContadorDeOcorrencia TweetsFiltrados" + INDICE_CONTAGEM + ".txt arquivoOcorrencias" + INDICE_CONTAGEM + ".txt\n").getBytes());
		ocorrenciaJDF.write(("final   : get arquivoOcorrencias" + INDICE_CONTAGEM + ".txt " + diretorioArquivos + "/Files/arquivoOcorrencias" + INDICE_CONTAGEM + ".txt").getBytes());
	}
	
	private static void executarJob(String job) throws Exception {
		imprime(Runtime.getRuntime().exec("bash broker addjob " + diretorioArquivos + "/Jobs/" + job + ".jdf", null, diretorioBroker));
	}
	
	private static boolean terminouJob(String job) throws Exception {
		Process p = Runtime.getRuntime().exec("bash broker status", null, diretorioBroker);
		BufferedReader reader = new BufferedReader(new InputStreamReader(p.getInputStream()));
		String linha = "";
		while ((linha = reader.readLine()) != null) {
			if (linha.contains(job)) {
				return linha.contains("[FINISHED]");
			}
		}
		return true;
	}
	
	private static boolean terminouTodosOsJobs() throws Exception {
		Process p = Runtime.getRuntime().exec("bash broker status", null, diretorioBroker);
		BufferedReader reader = new BufferedReader(new InputStreamReader(p.getInputStream()));
		String linha = "";
		while ((linha = reader.readLine()) != null) {
			if (linha.contains("[UNSTARTED]") || linha.contains("[RUNNING]")) {
				return false;
			}
		}
		return true;
	}
	
	private static void imprimeTempo() throws Exception {
		if (!terminouTodosOsJobs()) {
			imprimeTempo();
		} else {
			System.out.println("Tempo total da aplicacao (em milisegundos):" + (new GregorianCalendar().getTimeInMillis() - inicio));
		}
	}
	
	private static boolean precisaRepetir() {
		for (int i = 0; i < NUMERO_DE_MAQUINAS; i++) {
			if (maquinas[i].equals("ContadorDeOcorrencia12")) {
				return false;
			}
		}
		return true;
	}
	
	private static void imprime(Process p) throws Exception {
		BufferedReader reader = new BufferedReader(new InputStreamReader(p.getInputStream()));
		String linha = "";
		while ((linha = reader.readLine()) != null) {
			System.out.println(linha);
		}
	}
}